package com.coffee.netty.connection;

import com.coffee.concurrent.CoffeeThreadFactory;
import com.coffee.netty.CommandHandler;
import com.coffee.netty.RpcCodec;
import com.coffee.netty.config.ConfigManager;
import com.coffee.netty.util.NettyEventLoopUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import lombok.NonNull;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * ConnectionFactory to create connection.
 *
 * @author songkui
 * @since 2024/7/31
 */
public class CoffeeConnectionFactory implements ConnectionFactory {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(CoffeeConnectionFactory.class);

    private static final EventLoopGroup workerGroup = NettyEventLoopUtil.newEventLoopGroup(Runtime
                    .getRuntime().availableProcessors() + 1,
            new CoffeeThreadFactory(
                    "zebra-client-worker", true));

    private long heartSecond;
    protected Bootstrap bootstrap;

    private RpcCodec codec;
    private ChannelHandler heartbeatHandler;

    private ConnectionCommandHandler commandHandler;
    private final ReconnectionHandler reconnectionHandler;

    public CoffeeConnectionFactory() {
        reconnectionHandler = new ReconnectionHandler();
    }

    @Override
    public void registerRpcCodec(@NonNull RpcCodec rpcCodec) {
        this.codec = rpcCodec;
    }

    @Override
    public void registerCommandHandler(@NonNull ConnectionCommandHandler commandHandler) {
        this.commandHandler = commandHandler;
    }

    @Override
    public void registerHeartbeatHandler(@NonNull ChannelHandler processor) {
        this.heartbeatHandler = processor;
    }

    @Override
    public void setHeartSecond(long heartTime) {
        this.heartSecond = 0 == heartTime ? ConfigManager.tcp_idle() : heartTime;
    }


    @Override
    public void init() {
        bootstrap = new Bootstrap();
        bootstrap.group(workerGroup).channel(NettyEventLoopUtil.getClientSocketChannelClass())
                .option(ChannelOption.TCP_NODELAY, ConfigManager.tcp_nodelay())
                .option(ChannelOption.SO_REUSEADDR, ConfigManager.tcp_so_reuseaddr())
                .option(ChannelOption.SO_KEEPALIVE, ConfigManager.tcp_so_keepalive())
                .option(ChannelOption.SO_SNDBUF, ConfigManager.tcp_so_sndbuf())
                .option(ChannelOption.SO_RCVBUF, ConfigManager.tcp_so_rcvbuf());

        // init netty write buffer water mark
        initWriteBufferWaterMark();

        // init byte buf allocator
        if (ConfigManager.netty_buffer_pooled()) {
            this.bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        } else {
            this.bootstrap.option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT);
        }

        final ConnectionRemotingHandler rpcHandler = new ConnectionRemotingHandler(commandHandler, reconnectionHandler, this);

        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel channel) {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(codec.packDecoder());
                pipeline.addLast("decoder", codec.decoder());
                pipeline.addLast("encoder", codec.encoder());
                pipeline.addLast("idleStateHandler", new IdleStateHandler(0, heartSecond, 0, TimeUnit.SECONDS));
                if (null != heartbeatHandler) {
                    pipeline.addLast("heartbeatHandler", heartbeatHandler);
                }
                pipeline.addLast("handler", rpcHandler);
            }
        });
    }

    /**
     * init netty write buffer water mark
     */
    private void initWriteBufferWaterMark() {
        int lowWaterMark = ConfigManager.netty_buffer_low_watermark();
        int highWaterMark = ConfigManager.netty_buffer_high_watermark();
        if (lowWaterMark > highWaterMark) {
            throw new IllegalArgumentException(
                    String
                            .format(
                                    "[client side] zebra netty high water mark {%s} should not be smaller than low water mark {%s} bytes)",
                                    highWaterMark, lowWaterMark));
        } else {
            logger.warn(
                    "[client side] zebra netty low water mark is {} bytes, high water mark is {} bytes",
                    lowWaterMark, highWaterMark);
        }
        this.bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
                lowWaterMark, highWaterMark));
    }

    @Override
    public void connect(Connection connection) throws Exception {
        int timeOut = Math.max(connection.getTimeOut(), 20000);
        String address = connection.getHost() + ":" + connection.getPort();
        if (logger.isDebugEnabled()) {
            logger.debug("connectTimeout of address [{}] is [{}].", address, timeOut);
        }
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeOut);
        ChannelFuture future = bootstrap.connect(new InetSocketAddress(connection.getHost(), connection.getPort()));
        future.awaitUninterruptibly();
        if (future.isSuccess()) {
            connection.successConnect(future.channel());
            if (ConnectionType.DisconnectReconnect.equals(connection.getConnectionType())) {
                commandHandler.disconnectReconnect(connection);
            } else {
                commandHandler.initialConnect(connection);
            }
        } else {
            String errMsg = "Create connection to " + address + " failure: "+ future.cause();
            logger.error(errMsg);
            if (connection.isReconnect()) {
                reconnectionHandler.reconnect(this, connection);
            } else {
                throw new Exception(errMsg, future.cause());
            }
        }

    }


}
